/**
 * 实现企业微信 MQTT 交互
 */
import { getLogger, RepeatChecker } from '@gnlab/utils'
import { MemStream, mmap, mmergeMap, mdistinct } from '@gnlab/utils/rx'
import type { Subscription, Observable } from 'rxjs'
import { Subject, firstValueFrom } from 'rxjs'
import { filter, map, first } from 'rxjs/operators'
import config from '@/config'
import type Connection from '@/lib/mqtt'
import type { RecvMessage, HookData, SendMessage, AnyObject } from './protocol'
import {
  ProtocolError,
  MessageType,
  parse,
  getRecvMessageType,
  makeHookMessage,
  makeManageMessage,
} from './protocol'

const logger = getLogger('services/mqtt/communicate')

class TopicBase {
  connection: Connection
  protected accountIds: MemStream<string[] | null>
  usable: MemStream<boolean> // 当前的可用状态
  prepared: Observable<unknown> // 初次进入可用状态时触发

  constructor(connection: Connection, accountIds: string[] | Promise<string[]>) {
    this.connection = connection

    this.accountIds = new MemStream<string[] | null>(null, subscriber => {
      if (Array.isArray(accountIds)) subscriber.next(accountIds)
      else void accountIds.then(ids => subscriber.next(ids))
    })

    this.usable = this.connection.connected.pipe(
      mmergeMap(connected =>
        this.accountIds.pipe(mmap(currAccountIds => connected && currAccountIds !== null)),
      ),
      mdistinct(),
    )

    this.prepared = this.usable.pipe(
      filter(usable => usable),
      first(),
    )
    this.prepared.subscribe(() => this.onPrepared())
  }

  protected onPrepared() {}
}

/**
 * 监听指定 topic 的各类消息
 */
export type { RecvMessage, HookData }

interface ObserveItem<T, Ext extends AnyObject = never> {
  type: MessageType
  message: RecvMessage<T, Ext>
  accountId: string
}

export interface HookObserveItem<T, Ext extends AnyObject = never> {
  type: number // hook 消息类型
  data: T // hook data 内容
  message: RecvMessage<HookData<T>, Ext> // 原消息
}

type Predicate<T> = (value: T, index: number) => boolean
function mayFilter<T>(predicate?: Predicate<T>) {
  return filter<T>(predicate ?? (() => true))
}

export class TopicObservables extends TopicBase {
  protected onPrepared() {
    this.subscribe()
  }

  get topics() {
    return this.accountIds.value
      ? this.accountIds.value.map(accountId => ({
          accountId,
          topicId: `${config.MQTT_RECV_PREFIX}${accountId}`,
        }))
      : null
  }

  // 可调用以下方法来取消监听、重新监听消息
  protected subscriptions: Subscription[] | null = null
  subscribe() {
    if (!this.topics) return void logger.error('topicId 尚未生成，无法初始化订阅')
    if (!this.subscriptions) {
      this.subscriptions = this.topics.map(({ accountId, topicId }) =>
        this.connection.topic(topicId).subscribe(this.onMessage.bind(this, accountId)),
      )
    }
  }
  unsubscribe() {
    if (this.subscriptions) {
      this.subscriptions.forEach(subscription => subscription.unsubscribe())
      this.subscriptions = null
    }
  }

  protected message = new Subject<ObserveItem<unknown, AnyObject>>()

  protected repeatChecker = new RepeatChecker<string>()
  protected onMessage(accountId: string, { payload }: { payload: string }) {
    // 生产环境下，避免个别消息过长而无法正常输出，截取其一部分
    const logPayload =
      !config.IN_DEVELOPMENT && payload.length > 6000
        ? payload.slice(0, 3000) + ' ... ' + payload.slice(payload.length - 3000, payload.length)
        : payload

    // TODO: 目前会推送过来很多用不上的 11028 消息，直接过滤以免影响性能（后期让后端直接不要推过来）
    if (payload.includes('"type":11028')) return

    // type 为 30000 的是“回执”消息
    if (payload.includes('"type":30000'))
      logger.debug(`got response message(${payload.length})`, accountId, logPayload)

    let message: RecvMessage<unknown>
    try {
      message = parse(payload, accountId)
    } catch (e) {
      if (e instanceof ProtocolError) {
        logger.error(e.message, e.error, payload, accountId)
      } else {
        logger.error('消息解析失败，未知错误', e, payload, accountId)
      }
      return
    }

    const { msgId, type } = message
    if (msgId && type && this.repeatChecker.isRepeat(`${msgId}-${type}`)) {
      // msgId 和 type 都一样才认为是重复消息，因为会有 msgId 一样的 30000 type 的回执消息
      logger.debug('发现重复消息', message)
      return
    }

    const item = { type: getRecvMessageType(message), message, accountId }
    this.message.next(item)
  }

  protected mapHookItem<T, Ext extends AnyObject = never>() {
    return map<ObserveItem<HookData<T>, Ext>, HookObserveItem<T, Ext>>(item => {
      const {
        data: { type, data },
      } = item.message
      return { type, data, message: item.message }
    })
  }

  // 推送所有消息
  all<T, Ext extends AnyObject = never>() {
    return this.message as Subject<ObserveItem<T, Ext>>
  }

  // 推送所有 manage 消息
  manageWildcard<T, Ext extends AnyObject = never>() {
    return this.all<T, Ext>().pipe(
      filter(item => item.type === MessageType.Manage),
      map(item => item.message),
    )
  }

  // 推送指定 type 的 manage 消息
  manage<T, Ext extends AnyObject = never>(type: number) {
    return this.manageWildcard<T, Ext>().pipe(filter(message => message.type === type))
  }

  // 推送所有 hook 消息，包括没有 wxid 的消息
  hookWildcard<T, Ext extends AnyObject = never>() {
    return this.all<HookData<T>, Ext>().pipe(
      filter(item => item.type === MessageType.Hook || item.type === MessageType.HookPlain),
      this.mapHookItem<T, Ext>(),
    )
  }

  // 推送没有 wxid 的 hook 消息
  hookPlain<T, Ext extends AnyObject = never>(type: number) {
    return this.all<HookData<T>, Ext>().pipe(
      filter(item => item.type === MessageType.HookPlain && item.message.data.type === type),
      this.mapHookItem<T, Ext>(),
    )
  }

  // 推送有 wxid 的 hook 消息
  hook<T, Ext extends AnyObject = never>(type: number) {
    return this.all<HookData<T>, Ext>().pipe(
      filter(item => item.type === MessageType.Hook && item.message.data.type === type),
      this.mapHookItem<T, Ext>(),
    )
  }

  // 只想获取某类消息的第一个结果时，可调用以下方法（注意：此类方法返回的是 promise）
  protected async once<T>(ob: Observable<T>, predicate?: Predicate<T>) {
    return firstValueFrom(ob.pipe(mayFilter(predicate)))
  }
  async allOnce<T, Ext extends AnyObject = never>(predicate?: Predicate<ObserveItem<T, Ext>>) {
    return this.once(this.all<T, Ext>(), predicate)
  }
  async manageWildcardOnce<T, Ext extends AnyObject = never>(
    predicate?: Predicate<RecvMessage<T, Ext>>,
  ) {
    return this.once(this.manageWildcard<T, Ext>(), predicate)
  }
  async manageOnce<T, Ext extends AnyObject = never>(
    type: number,
    predicate?: Predicate<RecvMessage<T, Ext>>,
  ) {
    return this.once(this.manage<T, Ext>(type), predicate)
  }
  async hookWildcardOnce<T, Ext extends AnyObject = never>(
    predicate?: Predicate<HookObserveItem<T, Ext>>,
  ) {
    return this.once(this.hookWildcard<T, Ext>(), predicate)
  }
  async hookPlainOnce<T, Ext extends AnyObject = never>(
    type: number,
    predicate?: Predicate<HookObserveItem<T, Ext>>,
  ) {
    return this.once(this.hookPlain<T, Ext>(type), predicate)
  }
  async hookOnce<T, Ext extends AnyObject = never>(
    type: number,
    predicate?: Predicate<HookObserveItem<T, Ext>>,
  ) {
    return this.once(this.hook<T, Ext>(type), predicate)
  }
}

/**
 * 向指定 topic 发送消息
 */
export class TopicSender extends TopicBase {
  makeTopicId(message: SendMessage<AnyObject>, accountId?: string) {
    if (accountId === undefined) accountId = this.accountIds.value?.[0] ?? ''
    return `${config.MQTT_SEND_PREFIX}${message.wxid || accountId}`
  }

  send(message: SendMessage<AnyObject>, accountId?: string) {
    if (!this.connection.connected.value)
      return void logger.error('MQTT 客户端未连接，无法发送消息', message)

    if (!this.accountIds.value)
      return void logger.error('topicId 尚未生成，无法发送 MQTT 消息', message)

    if (accountId === undefined) accountId = this.accountIds.value[0] ?? ''
    const payload = JSON.stringify({ ...message, userid: accountId })
    this.connection.publish(this.makeTopicId(message), payload)
  }

  // 发送企业微信 hook 消息
  sendHook(wxid: string, type: number, data: AnyObject) {
    const msg = makeHookMessage(wxid, type, data)
    this.send(msg)
    return msg
  }

  // 发送管理类消息
  sendManage(type: number, data: AnyObject = {}) {
    const msg = makeManageMessage(type, data)
    this.send(msg)
    return msg
  }
}

/**
 * 整合 TopicObservables 和 TopicSender
 */
export class TopicCommunicate {
  protected observables: TopicObservables
  protected sender: TopicSender

  constructor(
    public connection: Connection,
    private readonly accountIds: string[] | Promise<string[]>,
  ) {
    this.observables = new TopicObservables(connection, accountIds)
    this.sender = new TopicSender(connection, accountIds)
  }

  get usable() {
    return this.sender.usable
  }
  get prepared() {
    return this.sender.prepared
  }

  get recvTopics() {
    return this.observables.topics
  }

  subscribe() {
    this.observables.subscribe()
  }
  unsubscribe() {
    this.observables.unsubscribe()
  }

  all<T, Ext extends AnyObject = never>(...args: Parameters<TopicObservables['all']>) {
    return this.observables.all<T, Ext>(...args)
  }
  manageWildcard<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['manageWildcard']>
  ) {
    return this.observables.manageWildcard<T, Ext>(...args)
  }
  manage<T, Ext extends AnyObject = never>(...args: Parameters<TopicObservables['manage']>) {
    return this.observables.manage<T, Ext>(...args)
  }
  hookWildcard<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['hookWildcard']>
  ) {
    return this.observables.hookWildcard<T, Ext>(...args)
  }
  hookPlain<T, Ext extends AnyObject = never>(...args: Parameters<TopicObservables['hookPlain']>) {
    return this.observables.hookPlain<T, Ext>(...args)
  }
  hook<T, Ext extends AnyObject = never>(...args: Parameters<TopicObservables['hook']>) {
    return this.observables.hook<T, Ext>(...args)
  }

  async allOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['allOnce']>
  ) {
    return this.observables.allOnce<T, Ext>(...args)
  }
  async manageWildcardOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['manageWildcardOnce']>
  ) {
    return this.observables.manageWildcardOnce<T, Ext>(...args)
  }
  async manageOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['manageOnce']>
  ) {
    return this.observables.manageOnce<T, Ext>(...args)
  }
  async hookWildcardOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['hookWildcardOnce']>
  ) {
    return this.observables.hookWildcardOnce<T, Ext>(...args)
  }
  async hookPlainOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['hookPlainOnce']>
  ) {
    return this.observables.hookPlainOnce<T, Ext>(...args)
  }
  async hookOnce<T, Ext extends AnyObject = never>(
    ...args: Parameters<TopicObservables['hookOnce']>
  ) {
    return this.observables.hookOnce<T, Ext>(...args)
  }

  send(...args: Parameters<TopicSender['send']>) {
    this.sender.send(...args)
    return this
  }
  sendHook(...args: Parameters<TopicSender['sendHook']>) {
    return this.sender.sendHook(...args)
  }
  sendManage(...args: Parameters<TopicSender['sendManage']>) {
    return this.sender.sendManage(...args)
  }
}
